package com.lemo.demo.hadoop;

import java.io.IOException;
import java.util.StringTokenizer;
import java.util.regex.Pattern;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.IntWritable.Comparator;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.io.WritableComparable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

/**
 * Hello world!
 *
 */
public class MapReduceApp2 extends Configured implements Tool
{
    public static void main( String[] args ) throws Exception
    {
    	args = new String[]{ //
    			"/user/beifeng/tmp1" ,//
    			"/user/beifeng/output1"
    		};
    	Configuration conf = new Configuration();
    	int status = ToolRunner.run(conf, new MapReduceApp2(), args);
    	System.exit(status);
    }

	@Override
	public int run(String[] arg0) throws Exception {
		// TODO Auto-generated method stub
		Configuration conf = this.getConf();
		Job job = Job.getInstance(conf,this.getClass().getSimpleName());
		job.setJarByClass(this.getClass());
		
		Path inpath = new Path(arg0[0]);
		FileInputFormat.addInputPath(job, inpath);
		//map
		job.setMapperClass(MapperClass.class);
		job.setMapOutputKeyClass(IntWritable.class);
		job.setMapOutputValueClass(IntWritable.class);
		
		//reduce
		job.setNumReduceTasks(1);
		job.setReducerClass(ReducerClass.class);
		job.setOutputKeyClass(IntWritable.class);
		job.setOutputValueClass(IntWritable.class);
		
		
		Path outpath = new Path(arg0[1]);
		FileOutputFormat.setOutputPath(job, outpath);
		
		job.setSortComparatorClass(myDecreasingComparator.class);   
		
		boolean sucess = job.waitForCompletion(true);
		return sucess? 0 : 1 ;
	}
    
	public static class MapperClass extends Mapper<LongWritable, Text, IntWritable, IntWritable> {

		private IntWritable mapOutputValue = new IntWritable();
		private IntWritable mapOutputKey = new IntWritable();
		@Override
		protected void map(LongWritable key, Text value, Context context)
				throws IOException, InterruptedException {
			// TODO Auto-generated method stub
			
			String lineValue = value.toString();
			
			// split
			String[] strs = lineValue.split("\\s+") ;
			System.out.println("key:"+strs[0]);
			mapOutputKey.set(Integer.parseInt(strs[1]));
			mapOutputValue.set(Integer.parseInt(strs[0]));
			context.write(mapOutputKey, mapOutputValue);
		}
		
	}
	public static class ReducerClass extends Reducer<IntWritable, IntWritable, IntWritable, IntWritable>{

		@Override
		protected void reduce(IntWritable key, Iterable<IntWritable> values,
				Context context) throws IOException, InterruptedException {
			for(IntWritable value:values) {
				context.write(value,key);
			}
			
		}
		
	}
	
	public static class myDecreasingComparator extends Comparator {  
        @SuppressWarnings("rawtypes")  
        public int compare( WritableComparable a,WritableComparable b){  
            return -super.compare(a, b);  
        }  
        public int compare(byte[] b1, int s1, int l1, byte[] b2, int s2, int l2) {  
            return -super.compare(b1, s1, l1, b2, s2, l2);  
        }  
    }  
    
}

